package com.shujia.dwi

import com.shujia.common.SparkTool
import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import com.shujia.util.Geography

object DwiResRegnMergelocationMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 将计算两个经纬度之间的距离注册成UDF 才能够在SQL/DSL中使用
    val calculateLength: UserDefinedFunction = udf((lg: String, lat: String, lg2: String, lat2: String) => {
      Geography.calculateLength(lg.toDouble, lat.toDouble, lg2.toDouble, lat2.toDouble)
    })


    // 从Hive中读取ods层oidd的数据
    val oiddDF: DataFrame = spark.table("ods.ods_oidd")

    /**
     * 计算速度：速度=距离/时间
     * 1、计算时间
     * 2、计算距离
     */
    val mergeDF: DataFrame = oiddDF
      // 进行分区裁剪
      .where($"day_id" === day_id)
      // 去除重复数据
      .select($"mdn", $"start_time", $"county_id", $"longi", $"lati", $"bsid", $"grid_id")
      .distinct()
      // 提取开始时间、结束时间
      .withColumn("start_t", split($"start_time", ",")(1))
      .withColumn("end_t", split($"start_time", ",")(0))

      /**
       * 问题点：最终计算的时间差有负数
       * 原因：相邻两条数据的时间有交叠
       * 解决方法：
       * 1、同一个网格中存在相邻两条数据的时间有交叠：在时间线上进行聚类
       * 2、不同网格中存在相邻两条数据的时间有交叠：直接用当前数据的起始时间减去上一条数据的起始时间
       */
      // 按照用户“分组”，按照时间升序排列，取前一条数据的网格id
      .withColumn("last_grid", lag("grid_id", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 对比当前数据的网格id和上一条数据的网格id，如果相同则置为0，不同则置为1
      .withColumn("flag", when($"grid_id" === $"last_grid", 0).otherwise(1))
      // 对上一步得到的flag进行累加求和 得到grp
      .withColumn("grp", sum($"flag") over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 对上一步得到的grp进行分组，合并数据
      .groupBy($"mdn", $"grid_id", $"county_id", $"bsid", $"longi", $"lati", $"grp")
      .agg(min($"start_t") as "start_t", max($"end_t") as "end_t")
      // 获取上一条数据的开始时间
      .withColumn("last_start_t", lag($"start_t", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 计算时间差（单位：s）
      .withColumn("diff_time", unix_timestamp($"start_t", "yyyyMMddHHmmss") - unix_timestamp($"last_start_t", "yyyyMMddHHmmss"))
      // 获取上一条数据的经纬度
      .withColumn("last_lg", lag($"longi", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
      .withColumn("last_lat", lag($"lati", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 计算距离（单位：m）
      .withColumn("distance", when($"last_lg".isNull, 0).otherwise(calculateLength($"longi", $"lati", $"last_lg", $"last_lat")))
      // 计算速度 注意时间差如果为null需要单独处理 否则会被后面的where条件过滤掉
      .withColumn("speed", when($"diff_time".isNull, 0.0).otherwise(round($"distance" / $"diff_time", 3)))
      .where($"speed" <= 340)

    mergeDF
      // 整理数据
      .select(
        upper(md5(concat($"mdn", expr("'shujia'")))) as "mdn"
        , $"start_t" as "start_date"
        , $"end_t" as "end_date"
        , $"county_id", $"longi"
        , $"lati"
        , $"bsid"
        , $"grid_id")
      // 将结果保存
      .write
      .mode(SaveMode.Overwrite)
      .format("csv")
      .option("sep", ",")
      .save(s"/daas/motl/dwi/dwi_res_regn_mergelocation_msk_d/day_id=$day_id")

    // 增加分区
    spark
      .sql(
        s"""
           |alter table dwi.dwi_res_regn_mergelocation_msk_d add if not exists partition (day_id=$day_id)
           |""".stripMargin)

  }

  /**
   * spark-submit --master yarn-client --class com.shujia.dwi.DwiResRegnMergelocationMskDay --jars common-1.0.jar dwi-1.0.jar 20220815
   */

}
